package main

import (
	"bufio"
	"fmt"
	"log"
	"net"
	"strings"

	resp "cn.webank/redis-sync/internal"
	rdb "github.com/hdt3213/rdb/core"
	"github.com/hdt3213/rdb/helper"
	"github.com/hdt3213/rdb/parser"
	uuid "github.com/satori/go.uuid"
)

type Redis struct {
	host   string
	passwd string
	conn net.Conn
	respDecoder *resp.Decoder
}

type RedisSrc struct {
	Redis

	rdbDecoder  *rdb.Decoder
}

type RedisDst struct {
	Redis
}

func CreateRedisSrc(host string, passwd string) (*RedisSrc, error) {
	c, err := net.Dial("tcp", host)
	if err != nil {
		return nil, fmt.Errorf("dial %s err: %v\n", host, err)
	}

	r := bufio.NewReader(c)
	rds := &RedisSrc{
		Redis: Redis{
			host: host,
			passwd: passwd,
			conn: c,
			respDecoder: resp.NewDecoder(r),
		},
		rdbDecoder:  parser.NewDecoder(r),
	}
	if passwd != "" {
		if err := execCmd(&rds.Redis, []interface{}{"AUTH", passwd}); err != nil {
			return nil, err
		}
	}
	return rds, nil
}

func CreateRedisDst(host string, passwd string) (*RedisDst, error) {
	c, err := net.Dial("tcp", host)
	if err != nil {
		return nil, fmt.Errorf("dial %s err: %v\n", host, err)
	}

	rds := &RedisDst{
		Redis: Redis{
			host: host,
			passwd: passwd,
			conn: c,
			respDecoder: resp.NewDecoder(c),
		},
	}
	if passwd != "" {
		if err := execCmd(&rds.Redis, []interface{}{"AUTH", passwd}); err != nil {
			return nil, err
		}
	}
	return rds, nil
}

type SyncTask struct {
	id string
	src *RedisSrc
	dst *RedisDst

	repl int64
	queue chan []byte
}

func NewSyncTask(src *RedisSrc, dst *RedisDst) *SyncTask {
	task := &SyncTask{
		id: uuid.NewV1().String(),
		src: src,
		dst: dst,
		queue: make(chan []byte, 1024),
	}

	return task
}

// 开启同步
func (t *SyncTask) Start() error {
	if err := execCmd(&t.src.Redis, []interface{}{"PSYNC", "?", "-1"}); err != nil {
		return err
	}

	_, _ = t.src.respDecoder.ReadCommand()

	if err := t.src.rdbDecoder.Parse(func(o parser.RedisObject) bool {
		cmd := helper.ObjectToCmd(o)
		data := helper.CmdLinesToResp(cmd)
		t.queue <- data
		return true
	}); err != nil {
		log.Default().Printf("parse rdb err:%v\n", err)
	}


	return nil
}

func (t *SyncTask) Stop() {
	if err := t.src.conn.Close(); err != nil {
		log.Default().Printf("close src:%s err:%v\n", t.src.host, err)
	}

	// TODO flush 保存repl
	if err := t.dst.conn.Close(); err != nil {
		log.Default().Printf("close dst:%s err:%v\n", t.dst.host, err)
	}
}


func execCmd(rds *Redis, arr []interface{}) error {
	cmd, err := resp.EncodeArray(arr)
	if err != nil {
		return fmt.Errorf("encode cmd err:%v\n", err)
	}

	_, err = rds.conn.Write(cmd)
	if err != nil {
		return fmt.Errorf("write cmd to %s err:%v\n", rds.host, err)
	}

	if result, err := rds.respDecoder.ReadCommand(); err != nil {
		return fmt.Errorf("decode cmd from %s err:%v\n", rds.host, err)
	} else {
		str := make([]string, 0)
		for _, v := range result {
			str = append(str, string(v))
		}
		fmt.Println(strings.Join(str, " "))
		return nil
	}
}